package com.ibm.cps.kafka.listener;

import com.ibm.cps.forwarder.MessageListenerV2;
import com.ibm.cps.kafka.KafkaTopicConsumeThread;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.newmessage.AbstractDataSourceMetadata;
import com.ibm.cps.newmessage.DataSourceMetadataFactory;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;
import org.apache.log4j.Logger;
import org.apache.storm.guava.base.Throwables;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by gongxuan on 12/3/15.
 */
public class ForwarderDeleteDataSourceListenerV2 extends KafkaTopicConsumeThread {

    private String tenantid;
    private Logger logger;
    private Map<String, MessageListenerV2> ds2Listener;

    public ForwarderDeleteDataSourceListenerV2(String zookeeper, String groupid, String topicName, String tenantid,
                                             Map<String, MessageListenerV2> ds2Listener) {
        super(zookeeper, groupid, topicName);
        logger = Logger.getLogger(ForwarderDeleteDataSourceListener.class);
        this.ds2Listener = ds2Listener;
        this.tenantid = tenantid;
    }

    @Override
    public void process(byte[] message) throws CPSException {
        String input = new String(message);
        String type;
        try {
            logger.info("Receive metadata in ForwarderDeleteDataSourceListener: " + input);
            HashMap<String, String> key2Value = JSONUtility.getKeyValue(input);
            type = key2Value.get(MessageFields.TYPE);
            logger.info("Type is " + type + " tenantid is " + tenantid);
            if (DataSourceMetadataFactory.isDataSourceType(type)) {
                if (key2Value.get(MessageFields.TENANTID).equals(tenantid)) {
                    AbstractDataSourceMetadata dataSource = DataSourceMetadataFactory.parseDataSourceMetadata(input,
                            type);
                    if (ds2Listener.get(dataSource.getName()) == null) {
                        logger.warn("There is no data source " + dataSource.getName() + " to be deleted.");
                        return;
                    }
                    ds2Listener.get(dataSource.getName()).removeDataSource();
                    ds2Listener.remove(dataSource.getName());
                    logger.info("Data source " + dataSource.getName() + " is deleted.");
                }
            }
        } catch (CPSException e) {
            logger.error(Throwables.getStackTraceAsString(e));
        }
    }
}
